/**********************************************************
*****************发布端程序publisher.cpp********************
***********************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <cstring>
#include <chrono>
#include <random>
#include <thread>
#include <atomic>
#include <mutex>
#include <iomanip>
#include <algorithm>
#include <windows.h>

/* IDL_TypeSupport.h中包含所有依赖的头文件 */
#include "IDL_TypeSupport.h"
#include "md5.h"             // 自定义MD5实现

using namespace std;
using namespace std::chrono;

// 全局统计变量
atomic<unsigned long long> total_sent(0);       // 总发送字节
atomic<unsigned long long> total_latency(0);   // 总延迟(us)
atomic<bool> test_completed(false);            // 测试完成标志
mutex print_mutex;                            // 输出锁

static int publisher_shutdown(DomainParticipant* participant) {
    ReturnCode_t retcode;
    int status = 0;

    if (participant != nullptr) {
        // 删除所有实体（DataWriter、Topic等）
        retcode = participant->delete_contained_entities();
        if (retcode != RETCODE_OK) {
            fprintf(stderr, "删除实体失败: %d\n", retcode);
            status = -1;
        }

        // 删除参与者
        retcode = DomainParticipantFactory::get_instance()->delete_participant(participant);
        if (retcode != RETCODE_OK) {
            fprintf(stderr, "删除参与者失败: %d\n", retcode);
            status = -1;
        }
    }
   return status;
}
template<typename T>
const T& clamp(const T& value, const T& low, const T& high) {
    return (value < low) ? low : (value > high) ? high : value;
}

/* 时延测试响应监听器 */
class ResponseListener : public DataReaderListener {
public:
    explicit ResponseListener(DataWriter* writer) : writer_(writer) {}

    void on_data_available(DataReader* reader) override {
        TestDataDataReader* response_reader = TestDataDataReader::narrow(reader);
        TestDataSeq data_seq;
        SampleInfoSeq info_seq;

        ReturnCode_t retcode = response_reader->take(
            data_seq, info_seq,
            LENGTH_UNLIMITED,
            ANY_SAMPLE_STATE,
            ANY_VIEW_STATE,
            ANY_INSTANCE_STATE
        );

        for (int i = 0; i < data_seq.length(); ++i) {
            auto now = duration_cast<microseconds>(
                system_clock::now().time_since_epoch()
            ).count();
            total_latency += (now - data_seq[i].send_time);
        }
    }

private:
    DataWriter* writer_;
};

/* 主测试函数 */
extern "C" int publisher_main(int domainId, int test_mode, int data_size, int loop_count) 
{
    system("set DDS_INTERFACE=WLAN");  // 指定物理网卡名称
    
    // DDS实体声明
    DomainParticipant* participant = nullptr;
    Publisher* publisher = nullptr;
    Subscriber* subscriber = nullptr;
    Topic* topic = nullptr;
    DataWriter* writer = nullptr;
    DataReader* response_reader = nullptr;
    TestDataDataWriter* TestData_writer = nullptr;
    TestData* instance = nullptr;
    ResponseListener* listener = nullptr;

    /* 1. 创建域参与者 */
    participant = DomainParticipantFactory::get_instance()->create_participant(
        domainId, PARTICIPANT_QOS_DEFAULT, nullptr, STATUS_MASK_NONE);
    if (!participant) {
        cerr << "创建participant失败" << endl;
        return -1;
    }

    /* 2. 创建发布者/订阅者 */
    publisher = participant->create_publisher(PUBLISHER_QOS_DEFAULT, nullptr, STATUS_MASK_NONE);
    subscriber = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT, nullptr, STATUS_MASK_NONE);
    if (!publisher || !subscriber) {
        cerr << "创建publisher/subscriber失败" << endl;
        return publisher_shutdown(participant);
    }

    /* 3. 注册数据类型 */
    const char* type_name = TestDataTypeSupport::get_type_name();
    if (TestDataTypeSupport::register_type(participant, type_name) != RETCODE_OK) {
        cerr << "注册类型失败" << endl;
        return publisher_shutdown(participant);
    }

    /* 4. 创建主题 */
    topic = participant->create_topic(
        "Example TestData",
        type_name,
        TOPIC_QOS_DEFAULT,
        nullptr,
        STATUS_MASK_NONE
    );
    if (!topic) {
        cerr << "创建topic失败" << endl;
        return publisher_shutdown(participant);
    }

    /* 5. 创建DataWriter */
    writer = publisher->create_datawriter(topic, DATAWRITER_QOS_DEFAULT, nullptr, STATUS_MASK_NONE);
    TestData_writer = TestDataDataWriter::narrow(writer);
    if (!TestData_writer) {
        cerr << "创建DataWriter失败" << endl;
        return publisher_shutdown(participant);
    }

    /* 6. 时延测试需要响应监听器 */
    if (test_mode == 3) {
        listener = new ResponseListener(writer);
        response_reader = subscriber->create_datareader(
            topic,
            DATAREADER_QOS_DEFAULT,
            listener,
            STATUS_MASK_ALL
        );
        if (!response_reader) {
            cerr << "创建response reader失败" << endl;
            return publisher_shutdown(participant);
        }
    }

    /* 7. 创建数据样本 */
    instance = TestDataTypeSupport::create_data();
    data_size = max(1, data_size);  // 强制最小值为1
    instance->payload.length(data_size);  // 正确初始化payload长度

    /* 8. 主发送循环 */
    random_device rd;
    mt19937 gen(rd());
    uniform_int_distribution<unsigned short> dist(0, 255);

    auto start_time = high_resolution_clock::now();
    auto last_print = start_time;

    for (int count = 0; count < (test_mode == 3 ? loop_count * 2 : loop_count); ++count) {
        // 生成随机数据
        for (int i = 0; i < data_size; ++i) {
            instance->payload[i] = dist(gen);
        }

        // 计算MD5
        MD5 md5;
        md5.update(instance->payload.get_buffer(), data_size);
        string md5_str = md5.hexdigest();
        // 确保 MD5 字符串长度为 32 字符
        if (md5_str.size() < 32) {
            cerr << "MD5 校验值无效！" << endl;
            return -1;
        }
        strncpy(instance->md5, md5_str.c_str(), 31);
        instance->md5[31] = '\0';  // 强制终止符

        // 设置元数据
        instance->seq_num = count;
        instance->send_time = duration_cast<microseconds>(
            system_clock::now().time_since_epoch()
        ).count();

        // 发送数据
        ReturnCode_t retcode = TestData_writer->write(*instance, HANDLE_NIL);
        if (retcode == RETCODE_OK) {
            total_sent += data_size;
        }
        else {
            cerr << "写入失败: " << retcode << endl;
            // 立即终止测试，避免后续错误
            TestDataTypeSupport::delete_data(instance);
            return publisher_shutdown(participant);
        }

        // 吞吐量统计显示
        if (test_mode == 2) {
            auto now = high_resolution_clock::now();
            if (duration_cast<seconds>(now - last_print).count() >= 1) {
                lock_guard<mutex> lock(print_mutex);
                double mbps = (total_sent.load() * 8.0) / 1e6;
                cout << "[TX] 吞吐量: " << fixed << setprecision(2) << mbps << " Mbps" << endl;
                total_sent = 0;
                last_print = now;
            }
        }

        // 时延测试模式控制
        if (test_mode == 3) {
            this_thread::sleep_for(500us); // 控制发送速率
        }
    }

    /* 9. 结果输出 */
    if (test_mode == 3) {
        auto total_time = duration_cast<microseconds>(
            high_resolution_clock::now() - start_time
        ).count();
        cout << "[时延] 平均: "
            << total_latency.load() / (loop_count * 2.0)
            << " us" << endl;
    }

    /* 10. 资源清理 */
    TestDataTypeSupport::delete_data(instance);
    if (listener) delete listener;
    return publisher_shutdown(participant);
}

/* 程序入口 */
int main(int argc, char* argv[]) 
{
    system("chcp 65001");  // 设置控制台为 UTF-8
    SetConsoleOutputCP(65001);
    if (argc < 5) {
        cerr << "用法: " << argv[0] << " <域ID> <测试模式> <数据大小> <循环次数>\n"
            << "  测试模式: 1-功能 2-吞吐量 3-时延\n"
            << "  数据大小: 1-1048576 字节\n"
            << "  循环次数: 测试迭代次数\n";
        return -1;
    }

    // 参数解析
    const int domain_id = atoi(argv[1]);
    const int test_mode = clamp(atoi(argv[2]), 1, 3);
    const int data_size = clamp(atoi(argv[3]), 1, 1048576);
    const int loop_count = max(atoi(argv[4]), 1);

    return publisher_main(domain_id, test_mode, data_size, loop_count);
}